#include "sqlfs.h"
#include "pool.h"
#include "log.h"
#include "fusexmp.h"
#include <errno.h>
#include <mysql++/mysql++.h>
#include "table.h"
#include <stdio.h>
#include <unistd.h>
#include <assert.h>
#include "int2bin.h"
#include <fcntl.h>
#include <sys/time.h>
#include "monitor.h"
using namespace std;
using namespace mysqlpp;
#ifdef DEBUG
#define catchex(x){\
	LOG << ENDL;\
	try {(x);}\
	catch(Exception e){ LOG_ERR << e.what() << ENDL; }\
	catch(...){ LOG_ERR << "something went terribly wrong" << ENDL;}}
#else
#define catchex(x) (x)
#endif

#define is_deleted(x) deleted_tables.find(x) != deleted_tables.end()

SqlFs *SqlFs::instance = NULL;


bool SqlFs::tableExists(const char* table) {		
//	LOG << ENDL;
	DB_INIT;
    Connection* con = p->grab();
    DB_START;
    Query q = con->query();
    q << "SHOW TABLES LIKE " << quote << table;
    vector<Row> res;
	q.storein(res);
	DB_STOP;
    p->release(con);	
    return !res.empty();
}

int SqlFs::numTables() {
	DB_INIT;
    Connection* con = p->grab();
    Query q = con->query();
    DB_START;
    q << "SHOW TABLES";
    vector<Row> res;
    q.storein(res);
    DB_STOP;
	p->release(con);
    return res.size();
}
static inline bool is_csv(const char *path) {
	int len = strlen(path);
	return len > 4 && strcmp(path + len - 4, ".csv") == 0;
}
static inline void get_table(const char* path, string* s) {
	*s = path + 1;
	*s = s->substr(0, s->size()-4);
}

bool SqlFs::isSqlFsFile(const char* path) {
	if(is_csv(path)) {
		//LOG << "csv file" << ENDL;
		string table;
		get_table(path, &table);
		//LOG << "table = " << table << ENDL;
		return tableExists(table.c_str());
	}
	//LOG << "not a table" << ENDL;
	return false;
}
void SqlFs::getAbsPath(const char* path, string *ret) {
	//LOG << "rootDir = " << rootDir << ENDL;
	*ret += rootDir;
	if(path[0] != '/')
		*ret += "/";
	*ret += path;
	//LOG << "ret = " << ret << ENDL;
}
//getattr: not need to be perfect. Just make sure tables are files, database is a dir.
int SqlFs::getattr(const char* path, struct stat* stbuf) {	
    //LOG << "path = " << path << ENDL;
	if(!isSqlFsFile(path)) {
		string abs_path;
		getAbsPath(path, &abs_path);
		return xmp_getattr(abs_path.c_str(), stbuf);
	}
	memset(stbuf, 0, sizeof(struct stat));
    stbuf->st_uid = getuid();
    stbuf->st_gid = getgid();
	stbuf->st_nlink = 1;
	stbuf->st_blocks = 8;
    if(!strcmp(path,"/")) {        
		stbuf->st_nlink = 2;
        stbuf->st_size = 4096;
        stbuf->st_mode = S_IFDIR | 0755;
		return 0;
    }

    string table;
    get_table(path, &table);
    if(is_deleted(table))
    	return -ENOENT;
	stbuf->st_mode = S_IFREG | 0644;    
	stbuf->st_size = Table::get(table)->size();
    return 0;
}



//read the list of tables in this db
int SqlFs::readdir(const char *path, void *buf, fuse_fill_dir_t filler, off_t offset, fuse_file_info *fi) {
	DB_INIT;
	string abs_path;
	getAbsPath(path, &abs_path);
	if(strcmp(path,"/")) {
		return xmp_readdir(abs_path.c_str(), buf, filler, offset, fi);
	}
	//BUG: fuse seems to call readdir on the same directory until the readdir reports empty directory.
	//we therefore set offset to 1 after the first call, then if offset is 1 we know that we already handled this request.
	if(offset)
		return 0;
	//first fill with xmp_readdir.
	xmp_readdir(abs_path.c_str(), buf, filler, offset, fi);
	Connection* con = p->grab();
	DB_START;
	Query q = con->query();    
	q << "SHOW TABLES";
	UseQueryResult res = q.use();
	DB_STOP;
	while(true) {
		DB_START;
		Row row = res.fetch_row();
		DB_STOP;
		if(!row)
			break;
		string table = row[0].c_str();
		if(is_deleted(table)) {
			continue;
		}
		table += ".csv";
		if(filler(buf, table.c_str(), NULL, 1))
			break;
	}
	p->release(con);
	return 0;
}



//open and release: just to do some book keeping.
int SqlFs::open(const char *path, struct fuse_file_info *fi) {

	if(!isSqlFsFile(path)) {

		string abs_path;
		getAbsPath(path, &abs_path);
		return xmp_open(abs_path.c_str(), fi);
	}
	string s;
	get_table(path, &s);
	if(is_deleted(s))
		return -ENOENT;
	Table *table = Table::get(s);
	fi->fh = (uint64_t)table;

    return 0;
}
int SqlFs::release(const char *path, struct fuse_file_info *fi) {


	if(!isSqlFsFile(path)) {

		string abs_path;
		getAbsPath(path, &abs_path);
		return xmp_release(abs_path.c_str(), fi);
	}

	string s;
	get_table(path, &s);
//Table *table =
	Table::get(s);

//	size_t lastlen;
//	if((lastlen = table->last_line.size()) > 0 && table->last_line[lastlen-1] != '\n') {
//		table->write("\n", 1, table->size());
//	}
	//delete table;
    return 0;
}
int SqlFs::flush(const char* path, struct fuse_file_info* fi) {
	if(!isSqlFsFile(path)) {

		return 0;
	}
	//LOG << "fi->fh = " << fi->fh << ENDL;
	string s;
	get_table(path, &s);
	Table *table = Table::get(s);

	size_t lastlen;
	//if((lastlen = table->last_line.size()) > 0 && table->last_line[lastlen-1] != '\n') {
	if((lastlen = table->linebuf_size) > 0 && table->linebuf[lastlen-1] != '\n') {
		printf("ADDING A NEW LINE TO THE FILE\n");
		table->write("\n", 1, table->size());
	}
	return 0;
}
//heart and soul for this app: init, read and write
//init, scan the entire database, create the offset data structures.
void* SqlFs::init(struct fuse_conn_info* conn) {
	DB_INIT;
	struct timeval start, end;
	long mtime, seconds, useconds;
	gettimeofday(&start, NULL);
	DB_START;
	Connection* con = p->grab();
    Query q = con->query();
    q << "SHOW TABLES";
	UseQueryResult res = q.use();
	DB_STOP;
	while(true) {
		DB_START;
		Row row = res.fetch_row();
		DB_STOP;
		if(!row)
			break;
		Table::get(row[0].c_str());
	}
	p->release(con);
	gettimeofday(&end, NULL);
	seconds  = end.tv_sec  - start.tv_sec;
	useconds = end.tv_usec - start.tv_usec;

	mtime = ((seconds) * 1000 + useconds/1000.0) + 0.5;

	printf("Elapsed time: %ld milliseconds\n", mtime);
	return NULL;
}
//read using the data structures created during initialization
int SqlFs::read(const char *path, char *buf, size_t size, off_t offset,
	      struct fuse_file_info *fi) {


	if(!is_csv(path)) {

		string abs_path;
		getAbsPath(path, &abs_path);
		return xmp_read(abs_path.c_str(), buf, size, offset, fi);
	}
	Table *table = (Table*)fi->fh;
	return table->read(buf, size, offset);
}
//write or lazy-write stuff. allow only append writes.
int SqlFs::write(const char *path, const char *buf, size_t size,
	       off_t offset, struct fuse_file_info *fi) {
    //LOG << ENDL;
    //LOG << "flags = " << to_bin(fi->flags) << ENDL;
	if(!is_csv(path)) {

		string abs_path;
		getAbsPath(path, &abs_path);
		return xmp_write(abs_path.c_str(), buf, size, offset, fi);
	}
	Table *table = (Table*)fi->fh;
	//LOG << "table name = " << table->name << ENDL;
    table->write(buf, size, offset);
    return size;
}
//allow access for anyone
int SqlFs::access(const char* path, int mask) {

	if(!isSqlFsFile(path)) {

		string abs_path;
		getAbsPath(path, &abs_path);
		return xmp_access(abs_path.c_str(), mask);
	}
    return 0;
}

int SqlFs::truncate(const char *path, off_t size) {
	if(!isSqlFsFile(path)) {

		string abs_path;
		getAbsPath(path, &abs_path);
		return xmp_truncate(abs_path.c_str(), size);
	}
    string s;
    get_table(path, &s);
    Table* table = Table::get(s);

    table->truncate(size);
    return 0;
}

int SqlFs::rename(const char *from, const char *to) {
	DB_INIT;
	if(!isSqlFsFile(from) && !is_csv(to)) {
		string from_path, to_path;
		getAbsPath(from, &from_path);
		getAbsPath(to, &to_path);
		return xmp_rename(from_path.c_str(), to_path.c_str());
	}
	if(isSqlFsFile(from) && is_csv(to)) {

		string from_table, to_table;
		get_table(from, &from_table);
		get_table(to, &to_table);
		Connection* con = p->grab();
		{
			DB_START;
			Query q = con->query();
			q << "DROP TABLE IF EXISTS " << to_table;
			q.exec();
			q.reset();
			q << "RENAME TABLE " << from_table << " TO " << to_table;
			q.execute();
			DB_STOP;
			Table::get(from_table)->rename(to_table);
		}
		p->release(con);

	}
	if(!isSqlFsFile(from) && isSqlFsFile(to)) {
		string from_path;
		getAbsPath(from, &from_path);
		string to_table;
		get_table(to, &to_table);
		//assert(is_deleted(to_table));
		struct stat st;
		lstat(from_path.c_str(), &st);
		int fd = ::open(from_path.c_str(), O_RDONLY);
		size_t read_size = 0;
		char buf[4096];
		Table& table = *(Table::get(to_table));
		table.truncate();
		while(read_size < (size_t)st.st_size) {
			size_t num_bytes = pread(fd, buf, 4096, read_size);
			//LOG << "buf = " << string(buf, num_bytes) << ENDL;
			//LOG << "num_bytes = " << num_bytes << ENDL;
			table.write(buf, num_bytes, read_size);
			read_size += num_bytes;
		}
		::close(fd);
		deleted_tables.erase(to_table);
		::unlink(from_path.c_str());
	}
	if(isSqlFsFile(from) && !isSqlFsFile(to)) {
		string from_table, to_path;
		getAbsPath(to, &to_path);
		get_table(from, &from_table);
		int fd = ::open(to_path.c_str(), O_WRONLY);
		size_t read_size = 0;
		char buf[4096];
		Table& table = *(Table::get(from_table));
		size_t table_size = table.size();
		while(read_size < table_size) {
			size_t num_bytes = table.read(buf, 4096, read_size);
			size_t write_size = 0;
			while(write_size < num_bytes) {
				write_size += ::write(fd, buf + write_size, num_bytes-write_size);
			}
			read_size += num_bytes;
		}
		::close(fd);
		deleted_tables.insert(from_table);
		table.truncate();
	}
    return 0;
}

//stuff that is optional to implement: flush,init,destroy,opendir,releasedir. 
//Some of them might be useful to manage caching and metadata



int SqlFs::opendir(const char* path, struct fuse_file_info* fi) {

	return 0;
}
int SqlFs::releasedir(const char* path, struct fuse_file_info* fi) {

	return 0;
}

void SqlFs::destroy(void* conn) {

}

//The rest of the FUSE api is not implemented.

int SqlFs::mknod(const char *path, mode_t mode, dev_t rdev) {
	if(!isSqlFsFile(path)) {

		string abs_path;
		getAbsPath(path, &abs_path);
		return xmp_mknod(abs_path.c_str(), mode, rdev);
	}
	string table_name;
	get_table(path, &table_name);
	if(is_deleted(table_name)) {
		deleted_tables.erase(table_name);
		return 0;
	} else {
		LOG_ERR << "mknod called for previously non-existent table" << ENDL;
		return -EPERM;
	}
}

int SqlFs::mkdir(const char *path, mode_t mode) {
	if(!isSqlFsFile(path)) {

		string abs_path;
		getAbsPath(path, &abs_path);
		return xmp_mkdir(abs_path.c_str(), mode);
	}
    LOG_ERR << "mkdir not implemented" << ENDL;
    return -EPERM;
}

int SqlFs::unlink(const char *path) {
	if(!isSqlFsFile(path)) {

		string abs_path;
		getAbsPath(path, &abs_path);
		return xmp_unlink(abs_path.c_str());
	}

    string s;
    get_table(path, &s);
    Table& table = *(Table::get(s));
    table.truncate();
    deleted_tables.insert(s);
    return 0;
}

int SqlFs::rmdir(const char *path) {
	if(!isSqlFsFile(path)) {

		string abs_path;
		getAbsPath(path, &abs_path);
		return xmp_rmdir(abs_path.c_str());
	}

    return -EPERM;
}

int SqlFs::symlink(const char *from, const char *to) {
	if(!isSqlFsFile(from)) {

		string from_path, to_path;
		getAbsPath(from, &from_path);
		getAbsPath(to, &to_path);
		return xmp_symlink(from_path.c_str(), to_path.c_str());
	}

    return -EPERM;
}



int SqlFs::link(const char *from, const char *to) {
	if(!isSqlFsFile(from)) {

		string from_path, to_path;
		getAbsPath(from, &from_path);
		getAbsPath(to, &to_path);
		return xmp_link(from_path.c_str(), to_path.c_str());
	}
    LOG_ERR << "link not implemented" << ENDL;
    return -EPERM;
}

int SqlFs::chmod(const char *path, mode_t mode) {
	if(!isSqlFsFile(path)) {

		string abs_path;
		getAbsPath(path, &abs_path);
		return xmp_chmod(abs_path.c_str(), mode);
	}

    return 0;
}

int SqlFs::chown(const char *path, uid_t uid, gid_t gid) {
	if(!isSqlFsFile(path)) {

		string abs_path;
		getAbsPath(path, &abs_path);
		return xmp_chown(abs_path.c_str(), uid, gid);
	}
    //LOG_ERR << "chown not implemented" << ENDL;
    return 0;
}



int SqlFs::utimens(const char *path, const struct timespec ts[2]){
	if(!isSqlFsFile(path)) {

		string abs_path;
		getAbsPath(path, &abs_path);
		return xmp_utimens(abs_path.c_str(), ts);
	}

    return 0;
}



int SqlFs::statfs(const char *path, struct statvfs *stbuf) {
	if(!isSqlFsFile(path)) {

		string abs_path;
		getAbsPath(path, &abs_path);
		return xmp_statfs(abs_path.c_str(), stbuf);
	}

    return -EPERM;
}



int SqlFs::fsync(const char *path, int isdatasync, struct fuse_file_info *fi) {
	if(!isSqlFsFile(path)) {

		string abs_path;
		getAbsPath(path, &abs_path);
		return xmp_fsync(abs_path.c_str(), isdatasync, fi);
	}

    return -EPERM;
}

int SqlFs::setxattr(const char *path, const char *name, const char *value,
		  size_t size, int flags) {

    return -EPERM;
}
int SqlFs::getxattr(const char *path, const char *name, char *value, size_t size) {

    return -EPERM;
}

int SqlFs::listxattr(const char *path, char *list, size_t size) {

    return -EPERM;
}

int SqlFs::removexattr(const char *path, const char *name) {

    return -EPERM;
}


int SqlFs::fsyncdir(const char* path, int datasync, struct fuse_file_info* fi) {	

	return -EPERM;
}
int SqlFs::readlink(const char* path, char* target, size_t size) {
	if(!isSqlFsFile(path)) {

		string abs_path;
		getAbsPath(path, &abs_path);
		return xmp_readlink(abs_path.c_str(), target, size);
	}

	return -EPERM;
}
